use clap::Parser;
use futures_util::{SinkExt, StreamExt};
use http_body_util::Full;
use hyper::body::Bytes;
use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper::{Request, Response};
use std::convert::Infallible;
use std::net::SocketAddr;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpSocket, TcpStream};
use tokio_tungstenite::{accept_async, tungstenite::Error};
use tracing::{debug, info, instrument, Level};
use tracing_subscriber::{fmt, layer::SubscriberExt, util::SubscriberInitExt};
use tungstenite::Result;

async fn hello(_: Request<hyper::body::Incoming>) -> Result<Response<Full<Bytes>>, Infallible> {
    let mut res = Response::new(Full::new(Bytes::from("Hello, World!")));

    res.headers_mut().append("MyKey", "MYVAL".parse().unwrap());
    res.extensions_mut().insert("HelloWorld");
    Ok(res)
}

async fn accept_connection(peer: SocketAddr, stream: TcpStream) {
    if let Err(e) = handle_connection(peer, stream).await {
        match e {
            Error::ConnectionClosed | Error::Protocol(_) | Error::Utf8 => (),
            err => eprintln!("Error processing connection: {}", err),
        }
    }
}

async fn handle_connection(peer: SocketAddr, stream: TcpStream) -> Result<()> {
    let mut ws_stream = accept_async(stream).await.expect("Failed to accept");

    println!("New WebSocket connection: {}", peer);

    while let Some(msg) = ws_stream.next().await {
        let msg = msg?;
        println!("len = {}", msg.len());
        //if msg.is_text() || msg.is_binary() {
        //    ws_stream.send(msg).await?;
        //}
    }

    Ok(())
}

/// Simple program to greet a person
///   You can use --help to print the help information
#[derive(Parser, Debug)]
#[command(author, version, about, long_about = None, verbatim_doc_comment)]
struct Args {
    /// Name of the person to greet
    #[arg(short, long)]
    name: String,

    /// Number of times to greet
    /// Has a multi-line help
    #[arg(short = 'a', long, default_value_t = 1, verbatim_doc_comment)]
    count: u8,
}

#[tokio::main]
async fn main() {
    let args = Args::parse();

    tracing_subscriber::fmt().with_max_level(Level::INFO).init();

    let tcp_serv = {
        let addr = "0.0.0.0:9900".parse().unwrap();
        let socket = TcpSocket::new_v4().unwrap();
        socket.set_reuseaddr(true).unwrap();
        socket.bind(addr).unwrap();
        let listener = socket.listen(1024).unwrap();
        listener
    };

    //let tcp_serv = TcpListener::bind("0.0.0.0:9900").await.unwrap();

    let websock_serv = TcpListener::bind("0.0.0.0:9701").await.unwrap();

    let http_serv = TcpListener::bind("0.0.0.0:9702").await.unwrap();

    loop {
        tokio::select! {
            _ = async {
                while let Ok((mut _stream, _)) = tcp_serv.accept().await {
                    tokio::spawn(async move {
                        let mut buffer = vec![0; 1024];

                        loop {
                            let size = _stream.read(&mut buffer).await.unwrap();
                            if size == 0 {
                                println!("tcp stream read size: {}", size);
                                break;
                            }
                            info!("recv: {:?}", &buffer[0..size]);
                            let val = serde_json::from_slice::<serde_json::Value>(&buffer[0..size]);
                            match val {
                                Ok(val) => {
                                    println!("parse json ok");
                                },
                                Err(err) => {
                                    println!("parse json err: {}", err);
                                }
                            }
                        }
                    });
                }
            } => {}
            _ = async {
                while let Ok((stream, _)) = websock_serv.accept().await {
                    let peer = stream.peer_addr().expect("connected streams should have a peer address");
                    println!("Peer address: {}", peer);
                    tokio::spawn(accept_connection(peer, stream));
                }
            } => {}
            _ = async {
                while let Ok((stream, _)) = http_serv.accept().await {
                    tokio::spawn(async move {
                        if let Err(err) = http1::Builder::new()
                        .title_case_headers(true)
                        .serve_connection(stream, service_fn(hello))
                        .await
                        {
                            println!("Error serving connection: {:?}", err);
                        }
                    });
                }
            } => {}
        }
    }
}
